package org.yumeko.kafka.topic;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartition;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;

//创建主题：createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options)
//删除主题：deleteTopics(final Collection<String> topicNames, DeleteTopicsOptions options)
//列出所有主题：listTopics(final ListTopicsOptions options)
//查询主题：describeTopics(final Collection<String> topicNames, DescribeTopicsOptions options)
//查询集群信息:describeCluster(DescribeClusterOptions options)
//查询配置信息:describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions options)
//修改配置信息:alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options)
//修改副本的⽇志⽬录：alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment, final AlterReplicaLogDirsOptions options)
//查询节点的⽇志⽬录信息：describeLogDirs(Collection<Integer> brokers, DescribeLogDirsOptions options)
//查询副本的⽇志⽬录信息：describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas,DescribeReplicaLogDirsOptions options)
//增加分区：createPartitions(Map<String, NewPartitions> newPartitions, final CreatePartitionsOptions options)
public class AdminclientTopic {

    private KafkaAdminClient client;

    @Before
    public void before() {

        Map<String, Object> configs = new HashMap<>();
        configs.put("bootstrap.servers", "192.168.88.129:9092");
        configs.put("client.id", "admin_001");
        client = (KafkaAdminClient) KafkaAdminClient.create(configs);
    }

    @After
    public void after() {
        // 关闭admin客户端
        client.close();
    }

    //列出所有主题
    @Test
    public void testListTopics() throws ExecutionException, InterruptedException {
        // 列出主题
//        final ListTopicsResult listTopicsResult = client.listTopics();

        ListTopicsOptions options = new ListTopicsOptions();
        // 列出内部主题
        options.listInternal(true);
        // 设置请求超时时间，单位是毫秒
        options.timeoutMs(500);

        ListTopicsResult listTopicsResult = client.listTopics(options);

//        final Set<String> strings = listTopicsResult.names().get();
//
//        strings.forEach(name -> {
//            System.out.println(name);
//        });

        // 将请求变成同步的请求，直接获取结果
        final Collection<TopicListing> topicListings = listTopicsResult.listings().get();

        topicListings.forEach(new Consumer<TopicListing>() {
            @Override
            public void accept(TopicListing topicListing) {

                // 该主题是否是内部主题
                final boolean internal = topicListing.isInternal();
                // 该主题的名字
                final String name = topicListing.name();


                System.out.println("主题是否是内部主题：" + internal);
                System.out.println("主题的名字：" + name);
                System.out.println(topicListing);
                System.out.println("=====================================");
            }
        });
    }

    //查询节点的⽇志⽬录信息
    @Test
    public void testDescribeLogDirs() throws ExecutionException, InterruptedException {
        final DescribeLogDirsResult describeLogDirsResult = client.describeLogDirs(Collections.singleton(0));

        Map<Integer, Map<String, LogDirDescription>> integerMapMap = describeLogDirsResult.allDescriptions().get();

        integerMapMap.forEach((integer, stringLogDirInfoMap) -> {
            System.out.println("broker.id = " + integer);
            //log.dirs可以设置多个目录
            stringLogDirInfoMap.forEach((s, logDirDescription) -> {
                System.out.println("logdir = " + s);
                Map<TopicPartition, ReplicaInfo> topicPartitionReplicaInfoMap = logDirDescription.replicaInfos();

                topicPartitionReplicaInfoMap.forEach((topicPartition, replicaInfo) -> {
                    System.out.println("主题分区：" + topicPartition.partition());
                    System.out.println("主题：" + topicPartition.topic());
//                                final boolean isFuture = replicaInfo.isFuture;
//                                final long offsetLag = replicaInfo.offsetLag;
//                                final long size = replicaInfo.size;
                });
            });
        });
    }
}
